Làm thế nào để giải quyết Captcha trong CrewAI với tích hợp CapSolver

pip install crewai
pip install 'crewai[tools]'
pip install requests

import requests
import time
from crewai.tools import BaseTool
from typing import Type
from pydantic import BaseModel, Field

CAPSOLVER_API_KEY = "YOUR_CAPSOLVER_API_KEY"

class CaptchaSolverInput(BaseModel):
    """Cấu trúc đầu vào cho công cụ giải CAPTCHA."""
    website_url: str = Field(..., description="Địa chỉ URL của trang web có CAPTCHA")
    website_key: str = Field(..., description="Mã trang của CAPTCHA")
    captcha_type: str = Field(default="ReCaptchaV2TaskProxyLess", description="Loại CAPTCHA cần giải")

class CaptchaSolverTool(BaseTool):
    name: str = "captcha_solver"
    description: str = "Giải các thách thức CAPTCHA bằng API CapSolver. Hỗ trợ reCAPTCHA v2, v3, Turnstile và nhiều loại khác."
    args_schema: Type[BaseModel] = CaptchaSolverInput

    def _run(self, website_url: str, website_key: str, captcha_type: str = "ReCaptchaV2TaskProxyLess") -> str:
        # Tạo nhiệm vụ
        create_task_url = "https://api.capsolver.com/createTask"

        task_payload = {
            "clientKey": CAPSOLVER_API_KEY,
            "task": {
                "type": captcha_type,
                "websiteURL": website_url,
                "websiteKey": website_key
            }
        }

        response = requests.post(create_task_url, json=task_payload)
        result = response.json()

        if result.get("errorId") != 0:
            return f"Lỗi tạo nhiệm vụ: {result.get('errorDescription')}"

        task_id = result.get("taskId")

        # Kiểm tra kết quả
        get_result_url = "https://api.capsolver.com/getTaskResult"

        for _ in range(60):  # Tối đa 60 lần thử
            time.sleep(2)

            result_payload = {
                "clientKey": CAPSOLVER_API_KEY,
                "taskId": task_id
            }

            response = requests.post(get_result_url, json=result_payload)
            result = response.json()

            if result.get("status") == "ready":
                solution = result.get("solution", {})
                return solution.get("gRecaptchaResponse") or solution.get("token")
            elif result.get("status") == "failed":
                return f"Nhận diện thất bại: {result.get('errorDescription')}"

        return "Hết thời gian chờ đợi giải CAPTCHA"

import requests
import time
from crewai.tools import BaseTool
from typing import Type
from pydantic import BaseModel, Field

CAPSOLVER_API_KEY = "YOUR_CAPSOLVER_API_KEY"


class ReCaptchaV2Input(BaseModel):
    """Cấu trúc đầu vào cho công cụ giải reCAPTCHA v2."""
    website_url: str = Field(..., description="Địa chỉ URL của trang web có reCAPTCHA v2")
    website_key: str = Field(..., description="Mã trang của reCAPTCHA")


class ReCaptchaV2Tool(BaseTool):
    name: str = "recaptcha_v2_solver"
    description: str = "Giải các thách thức reCAPTCHA v2 bằng CapSolver"
    args_schema: Type[BaseModel] = ReCaptchaV2Input

    def _run(self, website_url: str, website_key: str) -> str:
        payload = {
            "clientKey": CAPSOLVER_API_KEY,
            "task": {
                "type": "ReCaptchaV2TaskProxyLess",
                "websiteURL": website_url,
                "websiteKey": website_key
            }
        }
        return self._solve_captcha(payload)

    def _solve_captcha(self, payload: dict) -> str:
        # Tạo nhiệm vụ
        response = requests.post("https://api.capsolver.com/createTask", json=payload)
        result = response.json()

        if result.get("errorId") != 0:
            return f"Lỗi: {result.get('errorDescription')}"

        task_id = result.get("taskId")

        # Kiểm tra kết quả
        for attempt in range(60):
            time.sleep(2)
            result = requests.post(
                "https://api.capsolver.com/getTaskResult",
                json={"clientKey": CAPSOLVER_API_KEY, "taskId": task_id}
            ).json()

            if result.get("status") == "ready":
                return result["solution"]["gRecaptchaResponse"]
            if result.get("status") == "failed":
                return f"Thất bại: {result.get('errorDescription')}"

        return "Hết thời gian chờ đợi giải pháp"

import requests
import time
from crewai.tools import BaseTool
from typing import Type
from pydantic import BaseModel, Field

CAPSOLVER_API_KEY = "YOUR_CAPSOLVER_API_KEY"


class ReCaptchaV3Input(BaseModel):
    """Cấu trúc đầu vào cho công cụ giải reCAPTCHA v3."""
    website_url: str = Field(..., description="Địa chỉ URL của trang web có reCAPTCHA v3")
    website_key: str = Field(..., description="Mã trang của reCAPTCHA")
    page_action: str = Field(default="submit", description="Tham số hành động cho reCAPTCHA v3")


class ReCaptchaV3Tool(BaseTool):
    name: str = "recaptcha_v3_solver"
    description: str = "Giải các thách thức reCAPTCHA v3 với xác minh dựa trên điểm số"
    args_schema: Type[BaseModel] = ReCaptchaV3Input

    def _run(
        self,
        website_url: str,
        website_key: str,
        page_action: str = "submit"
    ) -> str:
        payload = {
            "clientKey": CAPSOLVER_API_KEY,
            "task": {
                "type": "ReCaptchaV3TaskProxyLess",
                "websiteURL": website_url,
                "websiteKey": website_key,
                "pageAction": page_action
            }
        }

        # Tạo nhiệm vụ
        response = requests.post("https://api.capsolver.com/createTask", json=payload)
        result = response.json()

        if result.get("errorId") != 0:
            return f"Lỗi: {result.get('errorDescription')}"

        task_id = result.get("taskId")

        # Kiểm tra kết quả
        for attempt in range(60):
            time.sleep(2)
            result = requests.post(
                "https://api.capsolver.com/getTaskResult",
                json={"clientKey": CAPSOLVER_API_KEY, "taskId": task_id}
            ).json()

            if result.get("status") == "ready":
                return result["solution"]["gRecaptchaResponse"]
            if result.get("status") == "failed":
                return f"Thất bại: {result.get('errorDescription')}"

        return "Hết thời gian chờ đợi giải pháp"

import requests
import time
from crewai.tools import BaseTool
from typing import Type
from pydantic import BaseModel, Field

CAPSOLVER_API_KEY = "YOUR_CAPSOLVER_API_KEY"


class TurnstileInput(BaseModel):
    """Cấu trúc đầu vào cho công cụ giải Turnstile."""
    website_url: str = Field(..., description="Địa chỉ URL của trang web có Turnstile")
    website_key: str = Field(..., description="Mã trang của widget Turnstile")


class TurnstileTool(BaseTool):
    name: str = "turnstile_solver"
    description: str = "Giải các thách thức Cloudflare Turnstile"
    args_schema: Type[BaseModel] = TurnstileInput

    def _run(self, website_url: str, website_key: str) -> str:
        payload = {
            "clientKey": CAPSOLVER_API_KEY,
            "task": {
                "type": "AntiTurnstileTaskProxyLess",
                "websiteURL": website_url,
                "websiteKey": website_key
            }
        }

        # Tạo nhiệm vụ
        response = requests.post("https://api.capsolver.com/createTask", json=payload)
        result = response.json()

        if result.get("errorId") != 0:
            return f"Lỗi: {result.get('errorDescription')}"

        task_id = result.get("taskId")

        # Kiểm tra kết quả
        for attempt in range(60):
            time.sleep(2)
            result = requests.post(
                "https://api.capsolver.com/getTaskResult",
                json={"clientKey": CAPSOLVER_API_KEY, "taskId": task_id}
            ).json()

            if result.get("status") == "ready":
                return result["solution"]["token"]
            if result.get("status") == "failed":
                return f"Thất bại: {result.get('errorDescription')}"

        return "Hết thời gian chờ đợi giải pháp"

import requests
import time
from crewai.tools import BaseTool
from typing import Type
from pydantic import BaseModel, Field

CAPSOLVER_API_KEY = "YOUR_CAPSOLVER_API_KEY"


class CloudflareChallengeInput(BaseModel):
    """Cấu trúc đầu vào cho công cụ giải thách thức Cloudflare."""
    website_url: str = Field(..., description="Địa chỉ URL của trang được bảo vệ")
    proxy: str = Field(..., description="Proxy theo định dạng: http://user:pass@ip:port")


class CloudflareChallengeTool(BaseTool):
    name: str = "cloudflare_challenge_solver"
    description: str = "Giải các trang thách thức Cloudflare (5 giây)"
    args_schema: Type[BaseModel] = CloudflareChallengeInput

    def _run(self, website_url: str, proxy: str) -> dict:
        payload = {
            "clientKey": CAPSOLVER_API_KEY,
            "task": {
                "type": "AntiCloudflareTask",
                "websiteURL": website_url,
                "proxy": proxy
            }
        }

        # Tạo nhiệm vụ
        response = requests.post("https://api.capsolver.com/createTask", json=payload)
        result = response.json()

        if result.get("errorId") != 0:
            return f"Lỗi: {result.get('errorDescription')}"

        task_id = result.get("taskId")

        # Kiểm tra kết quả
        for attempt in range(60):
            time.sleep(3)
            result = requests.post(
                "https://api.capsolver.com/getTaskResult",
                json={"clientKey": CAPSOLVER_API_KEY, "taskId": task_id}
            ).json()

            if result.get("status") == "ready":
                return {
                    "cookies": result["solution"]["cookies"],
                    "user_agent": result["solution"]["userAgent"]
                }
            if result.get("status") == "failed":
                return f"Thất bại: {result.get('errorDescription')}"

        return "Hết thời gian chờ đợi giải pháp"

from selenium import webdriver
from selenium.webdriver.common.by import By

def submit_recaptcha_token(driver, token: str):
    """Chèn token reCAPTCHA và gửi"""
    # Làm textarea ẩn trở nên hiển thị và đặt token
    recaptcha_response = driver.find_element(By.ID, "g-recaptcha-response")
    driver.execute_script("arguments[0].style.display = 'block';", recaptcha_response)
    recaptcha_response.clear()
    recaptcha_response.send_keys(token)

    # Gửi biểu mẫu
    form = driver.find_element(By.TAG_NAME, "form")
    form.submit()

def submit_turnstile_token(driver, token: str):
    """Chèn token Turnstile và gửi"""
    # Đặt token vào trường ẩn
    turnstile_input = driver.find_element(By.NAME, "cf-turnstile-response")
    driver.execute_script("arguments[0].value = arguments[1];", turnstile_input, token)

    # Gửi biểu mẫu
    form = driver.find_element(By.TAG_NAME, "form")
    form.submit()

def use_cookies_and_user_agent(driver, cookies: dict, user_agent: str):
    """Sử dụng cookies và user agent từ CapSolver"""
    # Thêm cookies vào trình duyệt
    for cookie in cookies:
        driver.add_cookie(cookie)

    # Thiết lập user agent
    driver.execute_cdp_cmd("Network.setUserAgentOverride", {"userAgent": user_agent})

# Thiết lập các cookie từ giải pháp CapSolver
for cookie in cf_solution["cookies"]:
    session.cookies.set(cookie["name"], cookie["value"])

# Thiết lập user agent đã được sử dụng để giải
headers = {
    "User-Agent": cf_solution["user_agent"]
}

# Bây giờ bạn có thể truy cập trang được bảo vệ
response = session.get(url, headers=headers)
return response.text

### Ví dụ Trích xuất Toàn diện

```python
import requests
import time
from selenium import webdriver
from selenium.webdriver.common.by import By

CAPSOLVER_API_KEY = "YOUR_CAPSOLVER_API_KEY"


def giai_recaptcha(website_url: str, website_key: str) -> str:
    """Nhận token reCAPTCHA từ CapSolver"""
    payload = {
        "clientKey": CAPSOLVER_API_KEY,
        "task": {
            "type": "ReCaptchaV2TaskProxyLess",
            "websiteURL": website_url,
            "websiteKey": website_key
        }
    }

    response = requests.post("https://api.capsolver.com/createTask", json=payload)
    result = response.json()

    if result.get("errorId") != 0:
        raise Exception(f"Lỗi: {result.get('errorDescription')}")

    task_id = result.get("taskId")

    for _ in range(60):
        time.sleep(2)
        result = requests.post(
            "https://api.capsolver.com/getTaskResult",
            json={"clientKey": CAPSOLVER_API_KEY, "taskId": task_id}
        ).json()

        if result.get("status") == "ready":
            return result["solution"]["gRecaptchaResponse"]
        if result.get("status") == "failed":
            raise Exception(f"Thất bại: {result.get('errorDescription')}")

    raise Exception("Hết thời gian")


def trich_xuat_voi_recaptcha(target_url: str, site_key: str):
    """Quy trình đầy đủ: giải CAPTCHA → gửi → trích xuất"""
    driver = webdriver.Chrome()
    driver.get(target_url)

    try:
        # 1. Giải CAPTCHA
        token = giai_recaptcha(target_url, site_key)

        # 2. Chèn token
        recaptcha_response = driver.find_element(By.ID, "g-recaptcha-response")
        driver.execute_script("arguments[0].style.display = 'block';", recaptcha_response)
        recaptcha_response.clear()
        recaptcha_response.send_keys(token)

        # 3. Gửi biểu mẫu
        driver.find_element(By.TAG_NAME, "form").submit()

        # 4. Trích xuất nội dung
        time.sleep(3)  # Chờ trang tải
        return driver.page_source

    finally:
        driver.quit()

from selenium import webdriver
from selenium.webdriver.chrome.options import Options

# Cấu hình Chrome với mở rộng CapSolver
chrome_options = Options()
chrome_options.add_extension("path/to/capsolver-extension.crx")

driver = webdriver.Chrome(options=chrome_options)

# Mở rộng sẽ tự động xử lý CAPTCHA
driver.get("https://example.com/trang-bao-vet")

def giai_voi_thu_lai(self, payload: dict, so_lan_thu: int = 3) -> str:
    for lan_thu in range(so_lan_thu):
        try:
            ket_qua = self._giai_captcha(payload)
            if ket_qua:
                return ket_qua
        except Exception as e:
            if lan_thu == so_lan_thu - 1:
                raise
            time.sleep(2 ** lan_thu)  # Chờ theo cấp số nhân

def kiem_tra_so_du() -> float:
    response = requests.post(
        "https://api.capsolver.com/getBalance",
        json={"clientKey": CAPSOLVER_API_KEY}
    )
    return response.json().get("balance", 0)

from functools import lru_cache
from datetime import datetime, timedelta

cache_captcha = {}

def lay_token_da_luu(website_url: str, website_key: str) -> str:
    cache_key = f"{website_url}:{website_key}"

    if cache_key in cache_captcha:
        token, timestamp = cache_captcha[cache_key]
        if datetime.now() - timestamp < timedelta(minutes=2):
            return token

    # Giải CAPTCHA mới
    new_token = giai_captcha(website_url, website_key)
    cache_captcha[cache_key] = (new_token, datetime.now())
    return new_token